/*
 * Copyright Debezium Authors.
 *
 * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 */
package io.debezium.connector.sqlserver.sink.object;

import org.apache.kafka.connect.data.Struct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Description: SourceField class
 * @author douxin
 * @date 2022/10/28
 **/
public class SourceField implements Cloneable {
    private static final Logger LOGGER = LoggerFactory.getLogger(SourceField.class);

    /**
     * Source
     */
    public static final String SOURCE = "source";

    /**
     * Snapshot
     */
    public static final String SNAPSHOT = "snapshot";
    /**
     * Database
     */
    public static final String DATABASE = "db";

    /**
     * Table
     */

    public static final String SCHEMA = "schema";

    public static final String TABLE = "table";

    public static final String SEQUENCE = "sequence";

    public static final String CHANGE_LSN = "change_lsn";

    public static final String COMMIT_LSN = "commit_lsn";

    public static final String EVENT_SERIAL_NO = "event_serial_no";

    /**
     * Gtid
     */
    // public static final String GTID = "gtid";
    //
    // /**
    // * Last_committed
    // */
    // public static final String LAST_COMMITTED = "last_committed";
    //
    // /**
    // * Sequence_number
    // */

    //
    // /**
    // * Binlog file
    // */
    // public static final String FILE = "file";
    //
    // /**
    // * Binlog position
    // */
    // public static final String POSITION = "pos";

    private String snapshot;
    private String database;

    private String schema;
    private String table;
    // private String gtid;
    // private long lastCommitted;
    private String sequence;
    // private String file;
    // private long position;
    private String change_lsn;
    private String commit_lsn;
    private long event_serial_no;

    /**
     * Constructor
     *
     * @param Struct the value
     */
    public SourceField(Struct value) {
        if (value == null) {
            throw new IllegalArgumentException("value can't be null!");
        }
        Struct source = value.getStruct(SourceField.SOURCE);
        if (source == null) {
            throw new IllegalArgumentException("source can't be null!");
        }
        this.snapshot = source.getString(SourceField.SNAPSHOT);
        this.database = source.getString(SourceField.DATABASE);
        this.schema = source.getString(SourceField.SCHEMA);
        this.table = source.getString(SourceField.TABLE).toLowerCase();
        this.sequence = source.getString(SourceField.SEQUENCE);
        this.change_lsn = source.getString(SourceField.CHANGE_LSN);
        this.commit_lsn = source.getString(SourceField.COMMIT_LSN);
        this.event_serial_no = source.getInt64(SourceField.EVENT_SERIAL_NO);

    }

    /**
     * Gets snapshot
     *
     * @return String the snapshot
     */
    public String getSnapshot() {
        return snapshot;
    }

    /**
     * Sets snapshot
     *
     * @param String the snapshot
     */
    public void setSnapshot(String snapshot) {
        this.snapshot = snapshot;
    }

    /**
     * Gets database
     *
     * @return String the database
     */
    public String getDatabase() {
        return database;
    }

    /**
     * Sets database
     *
     * @param String the database
     */
    public void setDatabase(String database) {
        this.database = database;
    }

    public String getSchema() {
        return schema;
    }

    public void setSchema(String schema) {
        this.schema = schema;
    }

    /**
     * Gets table
     *
     * @return String the table
     */

    public String getTable() {
        return table.toLowerCase();
    }

    /**
     * Sets table
     *
     * @param String the table
     */
    public void setTable(String table) {
        this.table = table.toLowerCase();
    }

    public String getCommit_lsn() {
        return commit_lsn;
    }

    public void setCommit_lsn(String commit_lsn) {
        this.commit_lsn = commit_lsn;
    }

    public String getChange_lsn() {
        return change_lsn;
    }

    public void setChange_lsn(String change_lsn) {
        this.change_lsn = change_lsn;
    }

    public Long getEvent_serial_no() {
        return event_serial_no;
    }

    public void setEvent_serial_no(Long event_serial_no) {
        this.event_serial_no = event_serial_no;
    }

    /**
     * Gets sequence
     *
     * @return long the sequence
     */
    public String getSequence() {
        return sequence;
    }

    /**
     * Sets sequence number
     *
     * @param long the sequence number
     */
    public void setSequence(String sequence) {
        this.sequence = sequence;
    }

    /**
     * Is full data, debezium provide source database's full data but we needn't handle it
     *
     * @return boolean the isFullData
     */
    public boolean isFullData() {
        return "true".equals(snapshot) || "last".equals(snapshot);
    }

    @Override
    public String toString() {
        return "SourceField{" +
                "snapshot='" + snapshot + '\'' +
                ", database='" + database + '\'' +
                ", sequence=" + sequence +
                ", schema='" + schema + '\'' +
                ", table='" + table + '\'' +
                ", change_lsn=" + change_lsn +
                ", commit_lsn='" + commit_lsn + '\'' +
                ", event_serial_no=" + event_serial_no +
                '}';
    }

    @Override
    public SourceField clone() {
        try {
            return (SourceField) super.clone();
        }
        catch (CloneNotSupportedException exp) {
            LOGGER.error("Clone source field failed.", exp);
        }
        return null;
    }
}
